//
//  ZYGCDSocketManager.m
//  ZYWebImage
//
//  Created by wangzhipan on 2025/4/23.
//

#import "ZYGCDSocketManager.h"
#import <sys/socket.h>
#import <netinet/in.h>
#import <arpa/inet.h>
#import <netdb.h>
#import <fcntl.h>
#import <unistd.h>
#import <errno.h>
#import <netinet/tcp.h>  // Add this import for TCP_NODELAY
#import "ZYSocketRequest.h"

static NSString * const kSocketError = @"com.ZYGCDSocketManager.Error";

// 定义读写操作类型
typedef NS_ENUM(NSInteger, ZYSocketOperation) {
    ZYSocketOperationNone,
    ZYSocketOperationRead,
    ZYSocketOperationWrite
};


@interface ZYGCDSocketManager() {
    int _socketFD;   // socket句柄，创建一个socket时返回的标识
    
    BOOL _isManualDisconnect;
    NSUInteger _reconnectAttempts;
    
    // 缓冲区管理
    NSMutableData *_readBuffer;
    NSMutableData *_writeBuffer;
    
    // 操作队列
    NSMutableArray<ZYSocketRequest *> *_readQueue;
    NSMutableArray<ZYSocketRequest *> *_writeQueue;
    
    // 当前操作
    ZYSocketOperation _currentOperation;
    
    // 超时定时器
    dispatch_source_t _readTimeoutTimer;
    dispatch_source_t _writeTimeoutTimer;
    
    // GCD事件源
    dispatch_source_t _readSource;
    dispatch_source_t _writeSource;
    
    // 线程安全队列
    dispatch_queue_t _socketQueue;
}

@property (nonatomic, copy, readwrite) NSString *host;
@property (nonatomic, assign, readwrite) NSUInteger port;
@property (nonatomic, assign, readwrite) BOOL isConnecting;
@property (nonatomic, strong, nullable) dispatch_source_t heartTimer;

@property (nonatomic, copy, nullable) NSData *heartData;
@end

@implementation ZYGCDSocketManager

- (instancetype)initWithHost:(NSString *)host port:(NSUInteger)port
{
    if (self = [super init]) {
        self.host = host;
        self.port = port;
        _socketFD = -1;
        _isManualDisconnect = NO;
        _isConnecting = NO;
        _timeout = 30;
        _autoReconnect = YES;
        _maxReconnectAttempts = 5;
        _reconnectAttempts = 0;
        _reconnectDelay = 2;
        
        // 初始化缓冲区
        _readBuffer = [NSMutableData dataWithCapacity:65536]; // 64KB 初始容量
        _writeBuffer = [NSMutableData dataWithCapacity:65536];
        
        // 初始化队列
        _readQueue = [NSMutableArray array];
        _writeQueue = [NSMutableArray array];
        
        // 初始化操作状态
        _currentOperation = ZYSocketOperationNone;
        
        // 创建串行队列确保线程安全
        _socketQueue = dispatch_queue_create("com.ZYGCDSocketManager.Queue", DISPATCH_QUEUE_SERIAL);
    }
    return self;
}

- (void)dealloc
{
    [self disconnect];
}

#pragma mark - Public Methods

- (BOOL)connect:(NSError * _Nullable *)error
{
    if (_isConnecting) {
        return YES;
    }
    _isManualDisconnect = NO;
    
    // 获取地址信息, 支持ipv4 && ipv6双栈
    struct addrinfo hints, *res;
    memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_UNSPEC;  // 支持ipv4 和 ipv6
    hints.ai_socktype = SOCK_STREAM;
    
    int status = getaddrinfo([_host UTF8String],
                             [[NSString stringWithFormat:@"%lu", _port] UTF8String],
                             &hints,
                             &res);
    
    if (status != 0) {
        if (error) {
            *error = [NSError errorWithDomain:kSocketError
                                         code:ISocketErrorCodeHostResolveFaild
                                     userInfo:@{NSLocalizedDescriptionKey:
                                                [NSString stringWithFormat:@"Failed to resolve hostname: %s",
                                                 gai_strerror(status)]}];
        }
        return NO;
    }
    
    // 尝试连接到第一个可用地址
    struct addrinfo *fp = res;
    while (fp != NULL) {
        _socketFD = socket(fp->ai_family, fp->ai_socktype, fp->ai_protocol);
        if (_socketFD >= 0) {
            // 设置超时
            struct timeval tv;
            tv.tv_sec = (int)_timeout;
            tv.tv_usec = (int)((_timeout - (int)_timeout) * 1000000);
            setsockopt(_socketFD, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
            setsockopt(_socketFD, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
            
            // 设置TCP_NODELAY选项，禁用Nagle算法，提高实时性
            int optval = 1;
            setsockopt(_socketFD, IPPROTO_TCP, TCP_NODELAY, &optval, sizeof(optval));
            
            // 设置SO_KEEPALIVE选项，保持连接活跃
            optval = 1;
            setsockopt(_socketFD, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval));
            
            if (connect(_socketFD, fp->ai_addr, fp->ai_addrlen) == 0) {
                // 连接成功
                break;
            }
            close(_socketFD);
            _socketFD = -1;
        }
        fp = fp->ai_next;
    }
    
    freeaddrinfo(res);
    
    if (_socketFD < 0) {
        if (error) {
            *error = [NSError errorWithDomain:kSocketError
                                         code:ISocketErrorCodeConnectFaild
                                     userInfo:@{NSLocalizedDescriptionKey: @"Failed to connect to server"}];
        }
        return NO;
    }
    
    // 设置为非阻塞模式
    int flags = fcntl(_socketFD, F_GETFL, 0);
    fcntl(_socketFD, F_SETFL, flags | O_NONBLOCK);
    
    _isConnecting = YES;
    _reconnectAttempts = 0;
    
    // 清空缓冲区和队列
    dispatch_sync(_socketQueue, ^{
        [_readBuffer setLength:0];
        [_writeBuffer setLength:0];
        [_readQueue removeAllObjects];
        [_writeQueue removeAllObjects];
        _currentOperation = ZYSocketOperationNone;
    });
    
    // 设置GCD事件源
    [self p_setupDispatchSources];
    
    return YES;
}

- (void)asyncConnect
{
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        NSError *error = nil;
        BOOL isSuccess = [self connect:&error];
        dispatch_async(dispatch_get_main_queue(), ^{
            if (isSuccess && [self.delegate respondsToSelector:@selector(socketDidConnect:)]) {
                [self.delegate socketDidConnect:self];
            }
            else if (!isSuccess && [self.delegate respondsToSelector:@selector(socket:didFailError:)]) {
                [self.delegate socket:self didFailError:error];
            }
        });
    });
}

- (void)disconnect
{
    _isManualDisconnect = YES;
    _isConnecting = NO;
    
    // 先停止心跳
    [self stopHeartbeat];
    
    // 停止所有超时定时器
    [self p_cancelTimeoutTimers];
    
    // 清理GCD事件源
    [self p_cleanupDispatchSources];
    
    // 关闭socket连接
    if (_socketFD >= 0) {
        close(_socketFD);
        _socketFD = -1;
    }
    
    // 清空缓冲区和队列
    dispatch_sync(_socketQueue, ^{
        [_readBuffer setLength:0];
        [_writeBuffer setLength:0];
        [_readQueue removeAllObjects];
        [_writeQueue removeAllObjects];
        _currentOperation = ZYSocketOperationNone;
    });
    
    if ([NSThread isMainThread]) {
        if ([self.delegate respondsToSelector:@selector(socket:didDisconnectWithError:)]) {
            [self.delegate socket:self didDisconnectWithError:nil];
        }
    }
    else {
        dispatch_async(dispatch_get_main_queue(), ^{
            if ([self.delegate respondsToSelector:@selector(socket:didDisconnectWithError:)]) {
                [self.delegate socket:self didDisconnectWithError:nil];
            }
        });
    }
}

- (BOOL)sendData:(NSData *)data error:(NSError * _Nullable *)error
{
    return [self sendData:data timeout:_timeout error:error];
}

- (BOOL)sendData:(NSData *)data timeout:(NSTimeInterval)timeout error:(NSError * _Nullable *)error
{
    if (!_isConnecting || _socketFD < 0) {
        if (error) {
            *error = [NSError errorWithDomain:kSocketError
                                         code:ISocketErrorCodeSendFaild
                                     userInfo:@{NSLocalizedDescriptionKey: @"socket not connect"}];
        }
        return NO;
    }
    
    __block BOOL success = NO;
    dispatch_sync(_socketQueue, ^{
        // 将数据添加到写缓冲区
        [_writeBuffer appendData:data];
        
        // 尝试立即发送
        success = [self p_flushWriteBuffer:error];
    });
    
    if (success) {
        dispatch_async(dispatch_get_main_queue(), ^{
            if (self.delegate && [self.delegate respondsToSelector:@selector(socket:didSendData:)]) {
                [self.delegate socket:self didSendData:data];
            }
        });
    }
    
    return success;
}

- (void)asyncSendData:(NSData *)data
{
    [self asyncSendData:data timeout:_timeout completion:nil];
}

- (void)asyncSendData:(NSData *)data timeout:(NSTimeInterval)timeout completion:(void (^)(NSError *error))completion
{
    if (!_isConnecting || _socketFD < 0) {
        NSError *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeSendFaild
                                         userInfo:@{NSLocalizedDescriptionKey: @"socket not connect"}];
        
        if (completion) {
            completion(error);
        } else {
            dispatch_async(dispatch_get_main_queue(), ^{
                if ([self.delegate respondsToSelector:@selector(socket:didFailError:)]) {
                    [self.delegate socket:self didFailError:error];
                }
            });
        }
        return;
    }
    
    NSData *sendData = [data copy];  // 防止数据被改、在其他线程被释放
    
    dispatch_async(_socketQueue, ^{
        // 创建写请求
        ZYSocketRequest *request = [[ZYSocketRequest alloc] init];
        request.data = sendData;
        request.timeout = timeout;
        request.startTime = [NSDate date];
        request.bytesDone = 0;
        request.bytesTotal = sendData.length;
        request.completion = ^(NSData *data, NSError *error) {
            if (completion) {
                completion(error);
            } else if (error) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    if ([self.delegate respondsToSelector:@selector(socket:didFailError:)]) {
                        [self.delegate socket:self didFailError:error];
                    }
                });
            }
        };
        
        // 添加到写队列
        [self->_writeQueue addObject:request];
        
        // 开始写操作
        [self p_startWriteOperation];
    });
}

- (void)readDataToLength:(NSUInteger)length timeout:(NSTimeInterval)timeout completion:(void (^)(NSData *data, NSError *error))completion
{
    if (!_isConnecting || _socketFD < 0) {
        NSError *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeReceiveFild
                                         userInfo:@{NSLocalizedDescriptionKey: @"socket not connect"}];
        if (completion) {
            completion(nil, error);
        }
        return;
    }
    
    dispatch_async(_socketQueue, ^{
        // 检查缓冲区中是否已有足够数据
        if (self->_readBuffer.length >= length) {
            NSData *data = [self->_readBuffer subdataWithRange:NSMakeRange(0, length)];
            [self->_readBuffer replaceBytesInRange:NSMakeRange(0, length) withBytes:NULL length:0];
            
            if (completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    completion(data, nil);
                });
            }
            return;
        }
        
        // 创建读请求
        ZYSocketRequest *request = [[ZYSocketRequest alloc] init];
        request.bytesTotal = length;
        request.bytesDone = self->_readBuffer.length;
        request.timeout = timeout;
        request.startTime = [NSDate date];
        request.completion = completion;
        
        // 添加到读队列
        [self->_readQueue addObject:request];
        
        // 开始读操作
        [self p_startReadOperation];
    });
}

- (void)readDataToData:(NSData *)data timeout:(NSTimeInterval)timeout completion:(void (^)(NSData *data, NSError *error))completion
{
    if (!_isConnecting || _socketFD < 0) {
        NSError *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeReceiveFild
                                         userInfo:@{NSLocalizedDescriptionKey: @"socket not connect"}];
        if (completion) {
            completion(nil, error);
        }
        return;
    }
    
    dispatch_async(_socketQueue, ^{
        // 检查缓冲区中是否已有包含分隔符的数据
        NSRange range = [self->_readBuffer rangeOfData:data options:0 range:NSMakeRange(0, self->_readBuffer.length)];
        if (range.location != NSNotFound) {
            NSUInteger endLocation = range.location + range.length;
            NSData *resultData = [self->_readBuffer subdataWithRange:NSMakeRange(0, endLocation)];
            [self->_readBuffer replaceBytesInRange:NSMakeRange(0, endLocation) withBytes:NULL length:0];
            
            if (completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    completion(resultData, nil);
                });
            }
            return;
        }
        
        // 创建读请求
        ZYSocketRequest *request = [[ZYSocketRequest alloc] init];
        request.data = data; // 存储分隔符
        request.timeout = timeout;
        request.startTime = [NSDate date];
        request.completion = completion;
        
        // 添加到读队列
        [self->_readQueue addObject:request];
        
        // 开始读操作
        [self p_startReadOperation];
    });
}

- (void)startHeartbeatWithData:(NSData *)data interval:(NSTimeInterval)interval
{
    [self stopHeartbeat];
    self.heartData = data;
    
    // 使用dispatch_source替代NSTimer
    self.heartTimer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, dispatch_get_main_queue());
    uint64_t intervalInNanoseconds = (uint64_t)(interval * NSEC_PER_SEC);
    dispatch_source_set_timer(self.heartTimer, dispatch_time(DISPATCH_TIME_NOW, intervalInNanoseconds),
                             intervalInNanoseconds, NSEC_PER_SEC / 10);
    
    __weak typeof(self) weakSelf = self;
    dispatch_source_set_event_handler(self.heartTimer, ^{
        [weakSelf p_sendHeartbeatData];
    });
    
    dispatch_resume(self.heartTimer);
}

- (void)stopHeartbeat
{
    if (self.heartTimer) {
        dispatch_source_cancel(self.heartTimer);
        self.heartTimer = nil;
    }
}

#pragma mark - Private Methods

// 设置GCD事件源
- (void)p_setupDispatchSources
{
    // 创建读事件源
    _readSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, _socketFD, 0, _socketQueue);
    
    __weak typeof(self) weakSelf = self;
    dispatch_source_set_event_handler(_readSource, ^{
        [weakSelf p_handleSocketReadEvent];
    });
    
    // 创建写事件源
    _writeSource = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, _socketFD, 0, _socketQueue);
    
    dispatch_source_set_event_handler(_writeSource, ^{
        [weakSelf p_handleSocketWriteEvent];
    });
    
    // 设置取消处理
    dispatch_source_set_cancel_handler(_readSource, ^{
        // 读事件源取消时的清理工作
    });
    
    dispatch_source_set_cancel_handler(_writeSource, ^{
        // 写事件源取消时的清理工作
    });
    
    // 启动读事件源
    dispatch_resume(_readSource);
    
    // 写事件源暂时不启动，等需要写数据时再启动
    // 注意：写事件源在创建后默认是挂起状态，需要在有数据要写时才resume
}

// 清理GCD事件源
- (void)p_cleanupDispatchSources
{
    if (_readSource) {
        dispatch_source_cancel(_readSource);
        _readSource = nil;
    }
    
    if (_writeSource) {
        dispatch_source_cancel(_writeSource);
        _writeSource = nil;
    }
}

// 处理读事件
- (void)p_handleSocketReadEvent
{
    if (!_isConnecting || _socketFD < 0) {
        return;
    }
    
    uint8_t buffer[16384]; // 16KB 缓冲区
    ssize_t readSize = recv(_socketFD, buffer, sizeof(buffer), 0);
    
    if (readSize > 0) {
        // 将读取的数据添加到缓冲区
        [_readBuffer appendBytes:buffer length:readSize];
        
        // 通知代理收到数据
        NSData *receivedData = [NSData dataWithBytes:buffer length:readSize];
        dispatch_async(dispatch_get_main_queue(), ^{
            if ([self.delegate respondsToSelector:@selector(socket:didReceiveData:)]) {
                [self.delegate socket:self didReceiveData:receivedData];
            }
        });
        
        // 处理读请求
        [self p_processReadBuffer];
    }
    else if (readSize == 0) {
        // 连接关闭
        NSError *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeConnectClosed
                                         userInfo:@{NSLocalizedDescriptionKey: @"ConnectClosed, Failed to read data"}];
        dispatch_async(dispatch_get_main_queue(), ^{
            [self p_handleDisconnect:error];
        });
    }
    else {
        // 读取错误
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // 非阻塞模式下，暂时没有数据可读，这是正常的
            return;
        }
        
        NSError *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeReceiveFild
                                         userInfo:@{NSLocalizedDescriptionKey: @"Failed to read data"}];
        dispatch_async(dispatch_get_main_queue(), ^{
            if ([self.delegate respondsToSelector:@selector(socket:didFailError:)]) {
                [self.delegate socket:self didFailError:error];
            }
        });
    }
}

// 处理写事件
- (void)p_handleSocketWriteEvent
{
    if (!_isConnecting || _socketFD < 0) {
        return;
    }
    
    // 尝试发送写缓冲区中的数据
    NSError *error = nil;
    [self p_flushWriteBuffer:&error];
    
    if (error) {
        // 发送失败
        if (error.code == ISocketErrorCodeConnectClosed) {
            dispatch_async(dispatch_get_main_queue(), ^{
                [self p_handleDisconnect:error];
            });
        } else {
            dispatch_async(dispatch_get_main_queue(), ^{
                if ([self.delegate respondsToSelector:@selector(socket:didFailError:)]) {
                    [self.delegate socket:self didFailError:error];
                }
            });
        }
    }
    
    // 如果写缓冲区为空，暂停写事件源
    if (_writeBuffer.length == 0 && _writeSource) {
        dispatch_suspend(_writeSource);
    }
}

// 处理读缓冲区
- (void)p_processReadBuffer
{
    // 没有读请求，直接返回
    if (_readQueue.count == 0) {
        return;
    }
    
    ZYSocketRequest *request = _readQueue[0];
    
    // 根据请求类型处理
    if (request.data) {
        // 读取直到特定分隔符
        NSRange range = [_readBuffer rangeOfData:request.data options:0 range:NSMakeRange(0, _readBuffer.length)];
        if (range.location != NSNotFound) {
            NSUInteger endLocation = range.location + range.length;
            NSData *resultData = [_readBuffer subdataWithRange:NSMakeRange(0, endLocation)];
            [_readBuffer replaceBytesInRange:NSMakeRange(0, endLocation) withBytes:NULL length:0];
            
            // 移除请求
            [_readQueue removeObjectAtIndex:0];
            
            // 取消超时定时器
            [self p_cancelReadTimeoutTimer];
            
            // 回调
            if (request.completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    request.completion(resultData, nil);
                });
            }
            
            // 处理下一个请求
            if (_readQueue.count > 0) {
                // 先重置操作状态，再开始下一个读操作
                _currentOperation = ZYSocketOperationNone;
                [self p_startReadOperation];
            } else {
                _currentOperation = ZYSocketOperationNone;
            }
        }
    } else {
        // 读取指定长度
        if (_readBuffer.length >= request.bytesTotal) {
            NSData *resultData = [_readBuffer subdataWithRange:NSMakeRange(0, request.bytesTotal)];
            [_readBuffer replaceBytesInRange:NSMakeRange(0, request.bytesTotal) withBytes:NULL length:0];
            
            // 移除请求
            [_readQueue removeObjectAtIndex:0];
            
            // 取消超时定时器
            [self p_cancelReadTimeoutTimer];
            
            // 回调
            if (request.completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    request.completion(resultData, nil);
                });
            }
            
            // 处理下一个请求
            if (_readQueue.count > 0) {
                // 先重置操作状态，再开始下一个读操作
                _currentOperation = ZYSocketOperationNone;
                [self p_startReadOperation];
            } else {
                _currentOperation = ZYSocketOperationNone;
            }
        }
    }
}

// 发送写缓冲区数据
- (BOOL)p_flushWriteBuffer:(NSError **)error
{
    if (_writeBuffer.length == 0) {
        return YES;
    }
    
    const uint8_t *bytes = (const uint8_t *)_writeBuffer.bytes;
    size_t length = _writeBuffer.length;
    size_t sent = 0;
    
    while (sent < length) {
        ssize_t result = send(_socketFD, bytes + sent, length - sent, 0);
        if (result < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // 非阻塞模式下，暂时无法发送，保留剩余数据
                if (sent > 0) {
                    [_writeBuffer replaceBytesInRange:NSMakeRange(0, sent) withBytes:NULL length:0];
                }
                break;
            } else {
                // 发送错误
                if (error) {
                    *error = [NSError errorWithDomain:kSocketError
                                                 code:ISocketErrorCodeSendFaild
                                             userInfo:@{NSLocalizedDescriptionKey: @"Failed to send data"}];
                }
                return NO;
            }
        } else if (result == 0) {
            // 连接已关闭
            if (error) {
                *error = [NSError errorWithDomain:kSocketError
                                             code:ISocketErrorCodeConnectClosed
                                         userInfo:@{NSLocalizedDescriptionKey: @"Connection closed by peer"}];
            }
            return NO;
        } else {
            // 发送成功，更新已发送字节数
            sent += result;
        }
    }
    
    // 更新写缓冲区
    if (sent > 0) {
        [_writeBuffer replaceBytesInRange:NSMakeRange(0, sent) withBytes:NULL length:0];
    }
    
    // 如果写队列中有请求，且已发送完所有数据，则完成请求
    if (_writeQueue.count > 0 && _writeBuffer.length == 0) {
        ZYSocketRequest *request = _writeQueue[0];
        [_writeQueue removeObjectAtIndex:0];
        
        // 取消超时定时器
        [self p_cancelWriteTimeoutTimer];
        
        // 回调
        if (request.completion) {
            dispatch_async(dispatch_get_main_queue(), ^{
                request.completion(request.data, nil);
            });
        }
        
        // 处理下一个请求
        if (_writeQueue.count > 0) {
            // 先重置操作状态，再开始下一个写操作
            _currentOperation = ZYSocketOperationNone;
            [self p_startWriteOperation];
        } else {
            _currentOperation = ZYSocketOperationNone;
        }
    }
    
    return YES;
}

// 开始读操作
- (void)p_startReadOperation
{
    if (_currentOperation == ZYSocketOperationRead) {
        return;
    }
    
    _currentOperation = ZYSocketOperationRead;
    
    // 设置读超时定时器
    ZYSocketRequest *request = _readQueue[0];
    [self p_setupReadTimeoutTimerWithInterval:request.timeout];
    
    // 读操作已经通过读事件源自动处理，这里不需要额外操作
}

// 开始写操作
- (void)p_startWriteOperation
{
    if (_currentOperation == ZYSocketOperationWrite) {
        return;
    }
    
    _currentOperation = ZYSocketOperationWrite;
    
    // 设置写超时定时器
    ZYSocketRequest *request = _writeQueue[0];
    [self p_setupWriteTimeoutTimerWithInterval:request.timeout];
    
    // 将请求数据添加到写缓冲区
    [_writeBuffer appendData:request.data];
    
    // 启动写事件源
    if (_writeSource) {
        dispatch_resume(_writeSource);
    }
    
    // 尝试立即发送
    NSError *error = nil;
    if (![self p_flushWriteBuffer:&error]) {
        // 发送失败
        [_writeQueue removeObjectAtIndex:0];
        [self p_cancelWriteTimeoutTimer];
        
        if (request.completion) {
            dispatch_async(dispatch_get_main_queue(), ^{
                request.completion(nil, error);
            });
        }
        
        // 处理下一个请求
        if (_writeQueue.count > 0) {
            _currentOperation = ZYSocketOperationNone;
            [self p_startWriteOperation];
        } else {
            _currentOperation = ZYSocketOperationNone;
        }
    }
}

// 设置读超时定时器
- (void)p_setupReadTimeoutTimerWithInterval:(NSTimeInterval)interval
{
    [self p_cancelReadTimeoutTimer];
    
    _readTimeoutTimer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, _socketQueue);
    uint64_t intervalInNanoseconds = (uint64_t)(interval * NSEC_PER_SEC);
    dispatch_source_set_timer(_readTimeoutTimer, dispatch_time(DISPATCH_TIME_NOW, intervalInNanoseconds),
                             DISPATCH_TIME_FOREVER, NSEC_PER_SEC / 10);
    
    __weak typeof(self) weakSelf = self;
    dispatch_source_set_event_handler(_readTimeoutTimer, ^{
        [weakSelf p_handleReadTimeout];
    });
    
    dispatch_resume(_readTimeoutTimer);
}

// 设置写超时定时器
- (void)p_setupWriteTimeoutTimerWithInterval:(NSTimeInterval)interval
{
    [self p_cancelWriteTimeoutTimer];
    
    _writeTimeoutTimer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, _socketQueue);
    uint64_t intervalInNanoseconds = (uint64_t)(interval * NSEC_PER_SEC);
    dispatch_source_set_timer(_writeTimeoutTimer, dispatch_time(DISPATCH_TIME_NOW, intervalInNanoseconds),
                             DISPATCH_TIME_FOREVER, NSEC_PER_SEC / 10);
    
    __weak typeof(self) weakSelf = self;
    dispatch_source_set_event_handler(_writeTimeoutTimer, ^{
        [weakSelf p_handleWriteTimeout];
    });
    
    dispatch_resume(_writeTimeoutTimer);
}

// 取消读超时定时器
- (void)p_cancelReadTimeoutTimer
{
    if (_readTimeoutTimer) {
        dispatch_source_cancel(_readTimeoutTimer);
        _readTimeoutTimer = nil;
    }
}

// 取消写超时定时器
- (void)p_cancelWriteTimeoutTimer
{
    if (_writeTimeoutTimer) {
        dispatch_source_cancel(_writeTimeoutTimer);
        _writeTimeoutTimer = nil;
    }
}

// 取消所有超时定时器
- (void)p_cancelTimeoutTimers
{
    [self p_cancelReadTimeoutTimer];
    [self p_cancelWriteTimeoutTimer];
}

// 处理读超时
- (void)p_handleReadTimeout
{
    if (_readQueue.count == 0) {
        [self p_cancelReadTimeoutTimer];
        return;
    }
    
    ZYSocketRequest *request = _readQueue[0];
    [_readQueue removeObjectAtIndex:0];
    
    NSError *error = [NSError errorWithDomain:kSocketError
                                         code:ISocketErrorCodeSendTimeout
                                     userInfo:@{NSLocalizedDescriptionKey: @"Read operation timed out"}];
    
    if (request.completion) {
        dispatch_async(dispatch_get_main_queue(), ^{
            request.completion(nil, error);
        });
    }
    
    // 处理下一个请求
    if (_readQueue.count > 0) {
        _currentOperation = ZYSocketOperationNone;
        [self p_startReadOperation];
    } else {
        _currentOperation = ZYSocketOperationNone;
    }
}

// 处理写超时
- (void)p_handleWriteTimeout
{
    if (_writeQueue.count == 0) {
        [self p_cancelWriteTimeoutTimer];
        return;
    }
    
    ZYSocketRequest *request = _writeQueue[0];
    [_writeQueue removeObjectAtIndex:0];
    
    NSError *error = [NSError errorWithDomain:kSocketError
                                         code:ISocketErrorCodeSendTimeout
                                     userInfo:@{NSLocalizedDescriptionKey: @"Write operation timed out"}];
    
    if (request.completion) {
        dispatch_async(dispatch_get_main_queue(), ^{
            request.completion(nil, error);
        });
    }
    
    // 处理下一个请求
    if (_writeQueue.count > 0) {
        _currentOperation = ZYSocketOperationNone;
        [self p_startWriteOperation];
    } else {
        _currentOperation = ZYSocketOperationNone;
    }
}

// 处理断开连接
- (void)p_handleDisconnect:(NSError *)error
{
    if (!_isConnecting) {
        return;
    }
    
    _isConnecting = NO;
    
    // 清理资源
    [self p_cleanupDispatchSources];
    
    if (_socketFD >= 0) {
        close(_socketFD);
        _socketFD = -1;
    }
    
    // 清空缓冲区和队列
    dispatch_sync(_socketQueue, ^{
        [_readBuffer setLength:0];
        [_writeBuffer setLength:0];
        
        // 处理未完成的读请求
        for (ZYSocketRequest *request in _readQueue) {
            if (request.completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    request.completion(nil, error);
                });
            }
        }
        [_readQueue removeAllObjects];
        
        // 处理未完成的写请求
        for (ZYSocketRequest *request in _writeQueue) {
            if (request.completion) {
                dispatch_async(dispatch_get_main_queue(), ^{
                    request.completion(nil, error);
                });
            }
        }
        [_writeQueue removeAllObjects];
        
        _currentOperation = ZYSocketOperationNone;
    });
    
    // 通知代理
    if ([self.delegate respondsToSelector:@selector(socket:didDisconnectWithError:)]) {
        [self.delegate socket:self didDisconnectWithError:error];
    }
    
    // 自动重连
    if (_autoReconnect && !_isManualDisconnect && _reconnectAttempts < _maxReconnectAttempts) {
        _reconnectAttempts++;
        
        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(_reconnectDelay * NSEC_PER_SEC)),
                      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
            [self asyncConnect];
        });
    }
}

// 发送心跳数据
- (void)p_sendHeartbeatData
{
    if (!_isConnecting || !self.heartData) {
        return;
    }
    
    [self asyncSendData:self.heartData];
}

@end
